
#include <apr-1/apr_general.h>
#include <apr-1/apr_getopt.h>
#include <apr-1/apr_strings.h>
#include "iface/APRInterface.h"
#include "iface/CryptInterface.h"
#include "iface/LibSSH2Interface.h"
#include "iface/SQLiteInterface.h"
#include "iface/XMLInterface.h"

#include "db/KeyDB.h"
#include "crypt/RSAKeyPair.h"
#include "net/TCPClient.h"

#include "net/Broker4.h"
#include "utils/Hash.h"
#include "utils/StringUtils.h"
#include "utils/SerialBlockSource.h"
#include "utils/SerialBlockSink.h"

#include "net/Broker4Client.h"

#include "xml/XMLDescriptor.h"

#define STANDARD_BLOCK 1048576

#include "acquire_client.h"

// Should consider using a special memory pool for returned objects!

typedef struct acquire_client_global_state
{
   APRInterface *apr_iface;
   SSLInterface *ssl_iface;
   CryptInterface *crypt_iface;
   LibSSH2Interface *ssh2_iface;
   SQLiteInterface *sqlite_iface;
   XMLInterface *xml_iface;
   KeyDB *keydb;
   const char *operational_brokerkeyname;
   const char *username;
   const char *password;
   const char *hostname;
   int portno;
   apr_pool_t *mp;
   int registration_state;
} AcquireClientGlobalState;

AcquireClientGlobalState state;

static int isInitialised = 0;

int AcquireClientStateIsIncomplete()
{
   int failed = 0x000;
   if (state.apr_iface == NULL)
   {
      errprint("the APR library has not been initialised");
      failed |= 0x0001;
   }
   if (state.ssl_iface == NULL)
   {
      errprint("openssl has not been initialised");
      failed |= 0x0002;
   }
   if (state.crypt_iface == NULL)
   {
      errprint("cryptlib has not been initialised");
      failed |= 0x0004;
   }
   if (state.ssh2_iface == NULL)
   {
      errprint("libssh2 has not been initialised");
      failed |= 0x0008;
   }
   if (state.sqlite_iface == NULL)
   {
      errprint("the sqlite library has not been initialised");
      failed |= 0x0010;
   }
   if (state.xml_iface == NULL)
   {
      errprint("libxml2 has not been initialised");
      failed |= 0x0020;
   }
   if (state.keydb == NULL)
   {
      errprint("the key database has not been initialised");
      failed |= 0x0040;
   }
   if (state.operational_brokerkeyname == NULL)
   {
      errprint("have not found a working broker key");
      failed |= 0x0080;
   }
   if (state.username == NULL)
   {
      errprint("username is not set");
      failed |= 0x0100;
   }
   if (state.password == NULL)
   {
      errprint("password is not set");
      failed |= 0x0200;
   }
   if (state.hostname == NULL)
   {
      errprint("remote (broker) hostname is not set");
      failed |= 0x0400;
   }
   if (state.portno == 0)
   {
      errprint("remote (broker) port number is not set");
      failed |= 0x0800;
   }
   if (state.mp == NULL)
   {
      errprint("results memory pool has not been initialised");
      failed |= 0x1000;
   }
   return failed;
}

int AcquireRegisterUserWithBroker()
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   char *pubkeyfingerprint = (state.keydb)->selfkeypair->GetMD5KeyFingerprint();
   printf("Registering key fingerprint %s as your key...\n", pubkeyfingerprint);
   free(pubkeyfingerprint);
   BrokerMessage *bmsg = new BrokerMessage(mp);
   bmsg->SetRecipient(state.operational_brokerkeyname);
   bmsg->SetSender(".%s", state.username);
   bmsg->AddTypeHeaderNOCRLF("REGISTER_USER_KEY", state.username);
   bmsg->AttachPublicKeyByName(state.keydb, (state.keydb)->selfkey);
   bmsg->AddVerbatimPayloadString((state.keydb)->WIPpassword);
   bmsg->AttachPublicKeyByName(state.keydb, "__SSH");
   bmsg->AttachPrivateKeyByName(state.keydb, "__SSH");
   bmsg->ForceNoSignature();
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   int retval = -1;
   if (bmsg->GetProcessee())
   {
      if (strcmp(bmsg->GetProcessee(), "AUTHORISED") == 0) retval = ACQUIRE_USER_REGISTER__USER_IS_AUTHORISED;
      if (strcmp(bmsg->GetProcessee(), "ACCEPTED") == 0) retval = ACQUIRE_USER_REGISTER__IS_ACCEPTED_FOR_REVIEW;
      if (strcmp(bmsg->GetProcessee(), "INVALID_NAME") == 0) retval = ACQUIRE_USER_REGISTER__NAME_IS_INVALID;
      if (strcmp(bmsg->GetProcessee(), "UNDER_REVIEW") == 0) retval = ACQUIRE_USER_REGISTER__USER_PENDING_REVIEW;
      if (strcmp(bmsg->GetProcessee(), "DENIED") == 0) retval = 0;
   }
   delete bmsg;
   apr_pool_destroy(mp);
   return retval;
}

int AcquireClientInit(const char *hostname, int portno, const char *username, const char *password, const char *brokername)
{
   AcquireClientInitLibraries();
   return AcquireClientInitState(hostname, portno, username, password, brokername);
}

void AcquireClientInitLibraries()
{
   state.apr_iface = new APRInterface();
   state.ssl_iface = new SSLInterface();
   state.crypt_iface = new CryptInterface();
   state.ssh2_iface = new LibSSH2Interface();
   state.sqlite_iface = new SQLiteInterface();
   state.xml_iface = new XMLInterface();
   state.mp = NULL;
   state.operational_brokerkeyname = NULL;
   state.username = NULL;
   state.hostname = NULL;
   state.portno = 0;
   state.keydb = NULL;
}

int AcquireClientInitState(const char *hostname, int portno, const char *username, const char *password, const char *brokername)
{
   state.portno = portno;
   // Someone keeps blanking my passwords!
   char *pass1 = strdup(password);
   char *pass2 = strdup(password);
   state.keydb = new KeyDB("client_keydb", password, username, password, 1);
   memset(pass1, 0, strlen(pass1));
   memset(pass2, 0, strlen(pass2));
   free(pass1);
   free(pass2);
   if (!((state.keydb)->HasOwnKey()))
   {
      errprint("could not find/generate own ('%s') key pair", username);
      delete state.keydb;
      state.keydb = NULL;
      return 0;
   }
   apr_status_t status;
   status = apr_pool_create(&(state.mp), (state.apr_iface)->getGlobalPool());
   CheckAPRError(status);
   if (state.mp == NULL)
   {
      errprint("memory pool for results not created");
      return 0;
   }
   (state.keydb)->ReplaceMainKeyPairWithUpdatedVersions(username);
   char *tmp_oper_brokerkeyname = (state.keydb)->GetLatestPublicKeyNameOfOwner(brokername);
   if ((tmp_oper_brokerkeyname == NULL) || (strlen(tmp_oper_brokerkeyname) == 0))
   {
      errprint("could not find any public keys owned by the broker");
      delete state.keydb;
      state.keydb = NULL;
      if (tmp_oper_brokerkeyname) free(tmp_oper_brokerkeyname);
      return 0;
   }
   char *passdup = strdup((state.keydb)->WIPpassword);
   if (!((state.keydb)->CheckForPublicKeyByName("__SSH")))
      (state.keydb)->GenerateKeyPairIntoDB(NULL, "__SSH", passdup, KEY_BITS);
   memset(passdup, 0, strlen(passdup));
   free(passdup);
   state.operational_brokerkeyname = tmp_oper_brokerkeyname;
   if ((hostname == NULL) || (strlen(hostname) == 0))
   {
      errprint("cannot connect to %s hostname", (hostname) ? "an empty" : "a NULL");
      return 0;
   }
   char *tmp_hostname = (char *)malloc(sizeof(char)*(strlen(hostname)+1));
   strcpy(tmp_hostname, hostname);
   state.hostname = tmp_hostname;
   if ((username == NULL) || (strlen(username) == 0))
   {
      errprint("username cannot be %s", (username) ? "left blank" : "NULL"); 
      return 0;
   }
   char *tmp_username = (char *)malloc(sizeof(char)*(strlen(username)+1));
   strcpy(tmp_username, username);
   state.username = tmp_username;
   if ((password == NULL) || (strlen(password) == 0))
   {
      errprint("password cannot be %s", (password) ? "left blank" : "NULL");
      return 0;
   }
   char *tmp_password = (char *)malloc(sizeof(char)*(strlen(password)+1));
   strcpy(tmp_password, password);
   state.password = tmp_password;
   int regis_t = AcquireRegisterUserWithBroker();
   switch (regis_t)
   {
      case -1: // broken connection / unknown response
      case 0: // access denied
      case ACQUIRE_USER_REGISTER__NAME_IS_INVALID:
         state.registration_state = ACQUIRE_USER_REGISTRATION_STATE__INVALID;
         break;
      case ACQUIRE_USER_REGISTER__USER_PENDING_REVIEW:
      case ACQUIRE_USER_REGISTER__IS_ACCEPTED_FOR_REVIEW:
         state.registration_state = ACQUIRE_USER_REGISTRATION_STATE__INCOMPLETE;
         break;
      case ACQUIRE_USER_REGISTER__USER_IS_AUTHORISED:
         state.registration_state = ACQUIRE_USER_REGISTRATION_STATE__FINAL;
         break;
   }
   if (state.registration_state == ACQUIRE_USER_REGISTRATION_STATE__FINAL)
   {
      isInitialised = 1;
   } else
   {
      AcquireClientDestroyState();
   }
   return state.registration_state;
}

void AcquireClientDestroyState()
{
   if (isInitialised == 0) return;
   isInitialised = 0;
   if (state.username) free((void *)state.username);
   state.username = NULL;
   if (state.password)
   {
      memset((void *)state.password, 0, strlen(state.password));
      free((void *)state.password);
   }
   state.password = NULL;
   if (state.hostname) free((void *)state.hostname);
   state.hostname = NULL;
   if (state.operational_brokerkeyname) free((void *)state.operational_brokerkeyname);
   state.operational_brokerkeyname = NULL;
   if (state.keydb) delete state.keydb;
   state.keydb = NULL;
   if (state.mp) apr_pool_destroy(state.mp);
   state.mp = NULL;
}

void AcquireClientDestroy()
{
   AcquireClientDestroyState();
   if (state.xml_iface) delete state.xml_iface;
   state.xml_iface = NULL;
   if (state.sqlite_iface) delete state.sqlite_iface;
   state.sqlite_iface = NULL;
   if (state.ssh2_iface) delete state.ssh2_iface;
   state.ssh2_iface = NULL;
   if (state.crypt_iface) delete state.crypt_iface;
   state.crypt_iface = NULL;
   if (state.ssl_iface) delete state.ssl_iface;
   state.ssl_iface = NULL;
   if (state.apr_iface) delete state.apr_iface;
   state.apr_iface = NULL;
}

void AcquireClientClearResults()
{
   if (AcquireClientStateIsIncomplete()) return;
   apr_pool_t *oldmp = state.mp;
   state.mp = NULL;
   apr_status_t status;
   status = apr_pool_create(&(state.mp), (state.apr_iface)->getGlobalPool());
   CheckAPRError(status);
   if (state.mp == NULL)
   {
      errprint("we do not have a memory pool for results");
      state.mp = oldmp;
   } else
   {
      apr_pool_destroy(oldmp);
   }
}

int AcquireUpdateKeys()
{
   // Return values may not make sense, needs further work...
   if (AcquireClientStateIsIncomplete()) return 0;
   // Can't call this before we are registered, the broker will not respond and keys won't update
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_key_name_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_key_name_of_user);
   bmsg->SetRecipient(state.operational_brokerkeyname);
   bmsg->AddTypeHeaderNOCRLF("REQUEST_UPDATE_KEYS", latest_key_name_of_user);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   if ((bmsg->IsType("KEY_UPDATE")) && (bmsg->IsPayloadCount(2)) && (strcmp(bmsg->GetSender(), state.operational_brokerkeyname) == 0))
   {
      StringUtils::aggressiveStringSanitiser((char *)bmsg->GetPayload(0));
      bmsg->TakeAndInsertOwnedPublicKeyByName(state.keydb, 1, bmsg->GetProcessee(), bmsg->GetPayload(0));
      (state.keydb)->DeletePublicKeyByName(state.operational_brokerkeyname);
      char *new_broker_key_uuid = (char *)malloc(sizeof(char)*(bmsg->GetPayloadLength(0)+1));
      strcpy(new_broker_key_uuid, bmsg->GetPayload(0));
      free((void *)(state.operational_brokerkeyname));
      state.operational_brokerkeyname = new_broker_key_uuid;
      // Instead of doing this on a timed basis, every time the broker key updates. we also update
      if ((state.keydb)->TryToUpdateOwnersKeys(NULL, state.username, 0))
      {
         // latest_keyname_of_user is now out of date, so
         char *old_keyname_of_user = latest_key_name_of_user;
         latest_key_name_of_user = NULL;
         bmsg->Reset();
         bmsg->SetSender(old_keyname_of_user);
         bmsg->SetRecipient(state.operational_brokerkeyname);
         bmsg->AddTypeHeaderNOCRLF("KEY_UPDATE", state.username);
         char *nlatest_key_name_of_user = (state.keydb)->GetLatestPublicKeyNameOfOwner(state.username);
         bmsg->AddVerbatimPayloadString(nlatest_key_name_of_user);
         bmsg->AttachPublicKeyByName(state.keydb, nlatest_key_name_of_user);
         Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
         broker4->MessageExchange(bmsg, state.keydb);
         delete broker4;
         if ((bmsg->IsType("KEYS_UPDATE_OKAY")) &&
             (strcmp(bmsg->GetSender(), state.operational_brokerkeyname) == 0) &&
             (strcmp(bmsg->GetRecipient(), nlatest_key_name_of_user) == 0))
         {
            if (strcmp(old_keyname_of_user, state.username) != 0)
            {
               (state.keydb)->DeletePublicKeyByName(old_keyname_of_user);
               (state.keydb)->DeletePrivateKeyByName(old_keyname_of_user);
            }
            printf("Key updates exchange successful!\n");
         } else if (bmsg->IsType("UPDATE_REFUSED"))
         {
            errprint("Key updates refused: untrusted username");
            errprint("rolling back keys to previous state");
            // Have to forget the key update, because the broker has not recieved it!
            (state.keydb)->RollbackOwnersKeyToPreviousKey(state.username, old_keyname_of_user);
         } else
         {
            errprint("do not understand reply (should be an acknowledgement of key exchange)");
            errprint("broker_uuid'%s'='%s', local_uuid'%s'='%s', messagetype'KEYS_UPDATE_OKAY'='%s'", 
                     state.operational_brokerkeyname, bmsg->GetSender(), nlatest_key_name_of_user, bmsg->GetRecipient(), bmsg->GetType());
            errprint("rolling back keys to previous state");
            // Have to forget the key update, because the broker has not recieved it!
            (state.keydb)->RollbackOwnersKeyToPreviousKey(state.username, old_keyname_of_user);
         }
         free(nlatest_key_name_of_user);
         free(old_keyname_of_user);
      } else
      {
         errprint("could not update user's local keys");
         if (latest_key_name_of_user) free(latest_key_name_of_user);
         apr_pool_destroy(mp);
         delete bmsg;      
         return 0;
      }
      delete bmsg;      
   } else if (bmsg->IsType("NO_UPDATE"))
   {
      printf("No key updates outstanding.\n");
      if (latest_key_name_of_user) free(latest_key_name_of_user);
      delete bmsg;
      apr_pool_destroy(mp);
      return 1;
   } else
   {
      errprint("unrecognised reply");
      if (latest_key_name_of_user) free(latest_key_name_of_user);
      delete bmsg;
      apr_pool_destroy(mp);
      return 0;
   }
   if (latest_key_name_of_user) free(latest_key_name_of_user);
   apr_pool_destroy(mp);
   return 1;
}

// Probably need to add number to spawn here too...
int AcquireGenerateInitialXMLWorkDescriptor(const char *descrfilename, const char *workname, const char *examplechain, const char *capabilities, int64_t maxtimesecs)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   XMLDescriptor *xmldescr = new XMLDescriptor();
   xmldescr->generateInitialDescriptor(workname, examplechain, capabilities, maxtimesecs);
   if (xmldescr->writePlain(descrfilename))
   {
      errprint("writing file '%s' failed", descrfilename);
      delete xmldescr;
      return 0;
   }
   delete xmldescr;
   return 1;
}

const char *AcquireReserveWorkStore(const char *descrfilename, const char *inputdirectory, int64_t *remotespace, int *blocks, int *cancel)
{
   if (AcquireClientStateIsIncomplete()) return NULL;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   RSAKeyPair *outbound_kp = new RSAKeyPair();
   outbound_kp->GenerateKeyPair(KEY_BITS);
   RSAKeyPair *inbound_kp = new RSAKeyPair();
   inbound_kp->GenerateKeyPair(KEY_BITS);
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(state.operational_brokerkeyname);
   
   bmsg->AddTypeHeaderNOCRLF("BEGIN_SESSION", state.username);
   bmsg->AddHeaderLineNOCRLF("Session-Type: WORK");
   bmsg->UpdateAndAttachXMLDescriptorByName(cancel, blocks, "unknown_key", outbound_kp, inputdirectory, descrfilename);
   if (cancel)
      if (*cancel)
      {
         delete bmsg;
         delete inbound_kp;
         delete outbound_kp;
         apr_pool_destroy(mp);
         cancel = 0;
         return NULL;
      }
   bmsg->AttachPublicKey(inbound_kp);
   bmsg->AttachPrivateKey(outbound_kp);
   bmsg->AttachPublicKey(outbound_kp);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   if (bmsg->IsPayloadCount(2))
   {
      const apr_table_t *ttable = bmsg->GetHeaderTable();
      StringUtils::aggressiveStringSanitiser((char *)bmsg->GetPayload(0));
      bmsg->TakeAndInsertPublicKeyByName(state.keydb, 1, bmsg->GetPayload(0));
      char *insess = StringUtils::varprintf("in_%s", bmsg->GetPayload(0));
      (state.keydb)->AddPrivateKeyWithPassword(insess, (state.keydb)->WIPpassword, (KeyPair *)inbound_kp, NULL);
      free(insess);
      char *outsess = StringUtils::varprintf("out_%s", bmsg->GetPayload(0));
      (state.keydb)->AddPrivateKeyWithPassword(outsess, (state.keydb)->WIPpassword, (KeyPair *)outbound_kp, NULL);
      (state.keydb)->AddPublicKey(outsess, (KeyPair *)outbound_kp, NULL);
      free(outsess);
      const char *stuffsize = apr_table_get(ttable, "SizeUsed");
      if (stuffsize)
      {
         apr_off_t ssize = 0;
         sscanf(stuffsize, "%" APR_OFF_T_FMT, &ssize);
         if (remotespace) *remotespace = ssize;
      }
   } else
   {
      delete bmsg;
      delete inbound_kp;
      delete outbound_kp;
      apr_pool_destroy(mp);
      return NULL;
   }
   char *retstr = StringUtils::mpprintf(state.mp, bmsg->GetPayload(0));
   delete bmsg;
   delete inbound_kp;
   delete outbound_kp;
   apr_pool_destroy(mp);
   if (cancel)
      if (*cancel)
      {
         AcquireDeleteWorkStore(retstr);
         free(retstr);
         cancel = 0;
         return NULL;
      }
   return retstr;
}

int AcquireDeleteWorkStore(const char *store_id)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(store_id);
   bmsg->AddTypeHeaderNOCRLF("EXPUNGE_SESSION", store_id);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   int retval = ACQUIRE_DELETE_WORK_STORE__UNKNOWN_ERROR;
   if ((bmsg->GetProcessee()) && (bmsg->GetType()))
   {
      if (strcmp(bmsg->GetProcessee(), "DENIED") == 0) retval = ACQUIRE_DELETE_WORK_STORE__ACCESS_DENIED;
      if (strcmp(bmsg->GetType(), "SESSION_EXPUNGED") == 0) retval = ACQUIRE_DELETE_WORK_STORE__SUCCESS;
      delete bmsg;
   }
   // Do local deletion from API databases
   if (retval == ACQUIRE_DELETE_WORK_STORE__SUCCESS)
   {
      (state.keydb)->DeletePublicKeyByName(store_id);
      (state.keydb)->DeletePrivateKeyByName(StringUtils::mpprintf(mp, "in_%s", store_id));
      (state.keydb)->DeletePublicKeyByName(StringUtils::mpprintf(mp, "out_%s", store_id));
      (state.keydb)->DeletePrivateKeyByName(StringUtils::mpprintf(mp, "out_%s", store_id));
   }
   apr_pool_destroy(mp);
   return retval;
}

int AcquireUploadToWorkStore(const char *store_id, const char *inputdirectory, const char *clusterid, const char *password, int instances, int maxtimesecs, int *blocksdone, int *cancelflag)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   int blocks_remaining = 1;
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   while (blocks_remaining > 0)
   {
      if (cancelflag)
         if (*cancelflag)
            break;
      int blockno = 0;
      SerialBlockSource *ses = new SerialBlockSource(inputdirectory, mp);
      int rsize = STANDARD_BLOCK;
      int blockcomplete = 0;
      while (rsize == STANDARD_BLOCK)
      {
         rsize = STANDARD_BLOCK;
         int cesize = 0;
         char *ceblock = ses->GetNextBlockEncrypted(StringUtils::mpprintf(mp, "out_%s", store_id), state.keydb, &rsize, &cesize, NULL);
         printf("CESIZE block %d is %d\n", blockno, cesize);
         BrokerMessage *bmsg = new BrokerMessage(mp);
         bmsg->SetRecipient(store_id);
         bmsg->SetSender(latest_keyname_of_user);
         bmsg->AddTypeHeaderNOCRLF("UPLOAD_OKAY", store_id);
         bmsg->AddHeaderLineNOCRLF("Block-Number: %d", blockno);
         Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
         broker4->MessageExchange(bmsg, state.keydb);
         delete broker4;
         bmsg->Print();
         if ((bmsg->GetType()) && (bmsg->GetProcessee()))
         {
            if ((strcmp(bmsg->GetType(), "FILECOMPLETE") == 0) &&
                  (strcmp(bmsg->GetProcessee(), "N") == 0))
            {
               bmsg->Reset();
               bmsg->SetRecipient(store_id);
               bmsg->SetSender(latest_keyname_of_user);
               bmsg->AddTypeHeaderNOCRLF("UPLOAD_DO", store_id);
               bmsg->AddHeaderLineNOCRLF("Block-Number: %d", blockno);
               char *phash = Hash::getSHA1Hash(ceblock, cesize);
               bmsg->AddVerbatimPayloadString(phash);
               bmsg->AddPayloadBuffer(ceblock, cesize);
               Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
               broker4->MessageExchange(bmsg, state.keydb);
               delete broker4;
               if ((strcmp(bmsg->GetType(), "FILECOMPLETE") == 0) &&
                   (strcmp(bmsg->GetProcessee(), "Y") == 0))
                      blockcomplete++;
               bmsg->Print();
            } else
            {
               free(ceblock);
            }
         }
         if (blocksdone)
            if ((*blocksdone) < blockcomplete)
               *blocksdone = blockcomplete;
         blockno++;
      }
      blocks_remaining = blockno - blockcomplete;
      delete ses;
   }
   if (blocks_remaining == 0)
   {
      BrokerMessage *bmsg = new BrokerMessage(mp);
      bmsg->SetRecipient(store_id);
      bmsg->SetSender(latest_keyname_of_user);
      bmsg->AddTypeHeaderNOCRLF("SET_ACTIVE", store_id);
      bmsg->AddHeaderLineNOCRLF("Max-Time: %d", maxtimesecs);
      bmsg->AddHeaderLineNOCRLF("Instances: %d", instances);
      if ((clusterid != NULL) && (strlen(clusterid) > 0))
         bmsg->AddHeaderLineNOCRLF("ClusterIDs: %s", clusterid);
      if (password)
         bmsg->AddVerbatimPayloadString(password);
      else
         bmsg->AddVerbatimPayloadString((state.keydb)->WIPpassword);
      Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
      broker4->MessageExchange(bmsg, state.keydb);
      delete broker4;
      bmsg->Print();
      free(latest_keyname_of_user);
      delete bmsg;
      apr_pool_destroy(mp);
      return 1;
   } else
   {
      free(latest_keyname_of_user);
      apr_pool_destroy(mp);
      return 0;
   }
}

int AcquireDownloadFromWorkStore(const char *store_id, const char *results_id, const char *outputdirectory, int *blocksdone, int *expected, int *cancelflag)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(store_id);
   bmsg->AddTypeHeaderNOCRLF("RESULTS_READY", store_id);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   const apr_table_t *aux_tbl = bmsg->GetAuxHeaderTable(0, "," CRLF_STR, ": ");
   const char *fullname = apr_table_get(aux_tbl, results_id);
   const char *blockno_str = apr_table_get(aux_tbl, fullname);
   if (blockno_str == NULL)
   {
      bmsg->Print();
      errprint("Invalid download (possibly wrong identifier '%s'?)", (results_id) ? results_id : "(null)");
      delete bmsg;
      apr_pool_destroy(mp);
      return 0;
   }
   int noofblocks = atoi(blockno_str);
   if (expected) *expected = noofblocks;
   bmsg->Reset();
   SerialBlockSink *ses = new SerialBlockSink(outputdirectory, mp);
   int blockno = 0;
   while (blockno < noofblocks)
   {
      if (cancelflag)
         if (*cancelflag)
            break;
      bmsg->SetRecipient(state.operational_brokerkeyname);
      char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
      bmsg->SetSender(latest_keyname_of_user);
      free(latest_keyname_of_user);
      bmsg->AddTypeHeaderNOCRLF("DOWNLOAD_DO", results_id);
      bmsg->AddHeaderLineNOCRLF("Block-Number: %d", blockno);
      broker4 = new Broker4Client(state.hostname, state.portno, mp);
      broker4->MessageExchange(bmsg, state.keydb);
      delete broker4;
      if (bmsg->IsPayloadCount(2))
      {
         StringUtils::aggressiveStringSanitiser((char *)bmsg->GetPayload(0));
         char *chash = Hash::getSHA1Hash(bmsg->GetPayload(1), bmsg->GetPayloadLength(1));
         if (strcmp(chash, bmsg->GetPayload(0)) == 0)
         {
            int realsize = 0;
            int cesize = bmsg->GetPayloadLength(1);
            printf("CESIZE of block %d bytes, hash %s\n", cesize, chash);
            char *hashval = NULL;
            ses->PutNextBlockDecrypted(StringUtils::mpprintf(mp, "in_%s", store_id), state.keydb, bmsg->GetPayload(1), &realsize, &cesize, &hashval);
            printf("BLOCK %d results - decoded size: %d hash: %s\n", blockno, realsize, hashval);
            blockno++;
            if (blocksdone)
               *blocksdone = blockno;
         } else errprint("BLOCK %d corrupt, trying again...", blockno);
      } else
      {
         errprint("download does not exist or is not ready");
         delete ses;
         delete bmsg;
         apr_pool_destroy(mp);
         return 0;
      }
      bmsg->Reset();
   }
   delete ses;
   delete bmsg;
   apr_pool_destroy(mp);
   return 1;
}

int AcquireAreResultsFromWorkStoreReady(const int nresultsarray, int *resultsarray, const char *store_id)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   for (int i = 0; i < nresultsarray; i++) resultsarray[i] = 0;
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(store_id);
   bmsg->AddTypeHeaderNOCRLF("RESULTS_READY", store_id);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   int retval = -1;
   if ((bmsg->GetProcessee()) && (bmsg->GetType()))
   {
      if (strcmp(bmsg->GetProcessee(), "DENIED") == 0) retval = -2;
      if ((strcmp(bmsg->GetType(), "RESULTS_DOWNLOADABLE") == 0) && (bmsg->IsPayloadCount(1)))
      {
         const apr_table_t *aux_tbl = bmsg->GetAuxHeaderTable(0, "," CRLF_STR, ": ");
         int resultno = 0;
         const char *val;
         while ((val = apr_table_get(aux_tbl, StringUtils::mpprintf(mp, "Result%d", resultno))) != NULL)
         {
            const char *workid = apr_table_get(aux_tbl, val);
            if (workid)
            {
               char *originator_copy = (char *)malloc(sizeof(char)*(strlen(workid)+1));
               strcpy(originator_copy, workid);
               char *remainder = NULL;
               char *ttoken = apr_strtok(originator_copy, "_", &remainder);
               ttoken = apr_strtok(NULL, "_", &remainder);
               if ((ttoken) && (atoi(ttoken) < nresultsarray))
               {
                  // Set the result in the array to TRUE
                  resultsarray[atoi(ttoken)] = 1;
               }
               free(originator_copy);
            }
            resultno++;
         }
         delete bmsg;
         apr_pool_destroy(mp);
         return 1;
      } else
      {
         delete bmsg;
         apr_pool_destroy(mp);
         return 0;
      }
   }
}

const char *AcquireGetResultIDFromNumberWhenReady(const int resultno, const char *store_id)
{
   if (AcquireClientStateIsIncomplete()) return NULL;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(store_id);
   bmsg->AddTypeHeaderNOCRLF("RESULTS_READY", store_id);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   int retval = -1;
   if ((bmsg->GetType()) && (bmsg->GetProcessee()))
   {
      if ((strcmp(bmsg->GetType(), "RESULTS_DOWNLOADABLE") == 0) && (bmsg->IsPayloadCount(1)))
      {
         const apr_table_t *aux_tbl = bmsg->GetAuxHeaderTable(0, "," CRLF_STR, ": ");
         int resultno = 0;
         const char *val;
         while ((val = apr_table_get(aux_tbl, StringUtils::mpprintf(mp, "Result%d", resultno))) != NULL)
         {
            const char *workid = apr_table_get(aux_tbl, val);
            if (workid)
            {
               char *originator_copy = (char *)malloc(sizeof(char)*(strlen(workid)+1));
               strcpy(originator_copy, workid);
               char *remainder = NULL;
               char *ttoken = apr_strtok(originator_copy, "_", &remainder);
               ttoken = apr_strtok(NULL, "_", &remainder);
               if ((ttoken) && (atoi(ttoken) == resultno))
               {
                  char *resultid_copy_palloc = (char *)apr_pstrdup(state.mp, val);
                  free(originator_copy);
                  delete bmsg;
                  apr_pool_destroy(mp);
                  return resultid_copy_palloc;
               }
               free(originator_copy);
            }
            resultno++;
         }
      }
   }
   delete bmsg;
   apr_pool_destroy(mp);
   return NULL;
}

int AcquireQueryWorkStatus(const char *store_id, float *pct_broker2compute, float *pct_complete, float *pct_compute2broker)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(state.operational_brokerkeyname);
   bmsg->AddTypeHeaderNOCRLF("QUERY_STATE", state.username);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   int retval = -2;
   if ((bmsg->GetProcessee()) && (bmsg->GetType()))
   {
      if (strcmp(bmsg->GetProcessee(), "DENIED") == 0) retval = 0;
      if ((strcmp(bmsg->GetType(), "ALLWORK") == 0) && (bmsg->IsPayloadCount(1)))
      {
         const apr_table_t *aux_tbl = bmsg->GetAuxHeaderTable(0, "," CRLF_STR, ": ");
         int resultno = 0;
         const char *val = apr_table_get(aux_tbl, store_id);
         if (val)
         {
            int pthou_broker2compute = 0;
            int pthou_complete = 0;
            int pthou_compute2broker = 0;
            sscanf(val, "PTHOU%d.%d.%d", &pthou_broker2compute, &pthou_complete, &pthou_compute2broker);
            if (pct_broker2compute) *pct_broker2compute = pthou_broker2compute / 10.;
            if (pct_complete) *pct_complete = pthou_complete / 10.;
            if (pct_compute2broker) *pct_compute2broker = pthou_compute2broker / 10.;
            retval = 1;
         } else
         {
            retval = -1;
         }
      }
      if ((strcmp(bmsg->GetType(), "ALLWORK") == 0) && (bmsg->IsPayloadCount(0)))
         retval = 2;
   }
   delete bmsg;
   apr_pool_destroy(mp);
   return retval;
}

int AcquireQueryAllWorkDataStatus(const char *store_id, char ***data_ids, char **stage_at, float **pct_complete, int *noofwds)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(state.operational_brokerkeyname);
   bmsg->AddTypeHeaderNOCRLF("QUERY_WORKDATA_STATE", store_id);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   int retval = ACQUIRE_QUERY_ALL_WORK__UNKNOWN_ERROR;
   if ((bmsg->GetProcessee()) && (bmsg->GetType()))
   {
      if (strcmp(bmsg->GetProcessee(), "DENIED") == 0) retval = ACQUIRE_QUERY_ALL_WORK__ACCESS_DENIED;
      if ((strcmp(bmsg->GetType(), "ALLWORKDATA") == 0) && (bmsg->IsPayloadCount(0)))
         retval = ACQUIRE_QUERY_ALL_WORK__SUCCESS_NO_WORK;
      if ((strcmp(bmsg->GetType(), "ALLWORKDATA") == 0) && (bmsg->IsPayloadCount(1)))
      {
         char *streamcopy = apr_pstrdup(mp, bmsg->GetPayload(0));
         const apr_table_t *aux_tbl = bmsg->GetAuxHeaderTable(0, "," CRLF_STR, ": ");
         int count = bmsg->GetAuxHeaderCount();
         if (data_ids) *data_ids = (char **)apr_palloc(state.mp, sizeof(char *)*count);
         if (stage_at) *stage_at = (char *)apr_palloc(state.mp, sizeof(char)*count);
         if (pct_complete) *pct_complete = (float *)apr_palloc(state.mp, sizeof(float)*count);
         int resultno = 0;
         const char *val;
         char *remainder = NULL;
         char *ttoken = apr_strtok(streamcopy, "," CRLF_STR, &remainder);
         while (ttoken != NULL)
         {
            if (resultno >= count)
            {
               errprint("broken: too many lines found");
               retval = ACQUIRE_QUERY_ALL_WORK__PARSE_ERROR;
               break;
            }
            char *remainder2 = NULL;
            char *workdata_id = apr_strtok(ttoken, ": ", &remainder2);
            const char *pthou_data = apr_table_get(aux_tbl, workdata_id);
            char workdata_stage = '\0';
            int workdata_pthou = 0;
            sscanf(pthou_data, "%c-%d", &workdata_stage, &workdata_pthou);
            if (data_ids) (*data_ids)[resultno] = (char *)apr_pstrdup(state.mp, workdata_id);
            if (stage_at) (*stage_at)[resultno] = workdata_stage;
            if (pct_complete) (*pct_complete)[resultno] = workdata_pthou / 10.;
            ttoken = apr_strtok(NULL, "," CRLF_STR, &remainder);
            resultno++;
         }
         if (noofwds) *noofwds = resultno;
         retval = ACQUIRE_QUERY_ALL_WORK__SUCCESS;
      }
   }
   delete bmsg;
   apr_pool_destroy(mp);
   return retval;
}

int AcquireQueryAllWorkStatus(char ***store_ids, float **pct_broker2compute, float **pct_complete, float **pct_compute2broker, int *noofstores, int64_t *quota)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(state.operational_brokerkeyname);
   bmsg->AddTypeHeaderNOCRLF("QUERY_STATE", state.username);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   int retval = ACQUIRE_QUERY_ALL_WORK__UNKNOWN_ERROR;
   if ((bmsg->GetProcessee()) && (bmsg->GetType()))
   {
      if (strcmp(bmsg->GetProcessee(), "DENIED") == 0) retval = ACQUIRE_QUERY_ALL_WORK__ACCESS_DENIED;
      if ((strcmp(bmsg->GetType(), "ALLWORK") == 0) && (bmsg->IsPayloadCount(0)))
         retval = ACQUIRE_QUERY_ALL_WORK__SUCCESS_NO_WORK;
      if ((strcmp(bmsg->GetType(), "ALLWORK") == 0) && (bmsg->IsPayloadCount(1)))
      {
         char *streamcopy = apr_pstrdup(mp, bmsg->GetPayload(0));
         const apr_table_t *aux_tbl = bmsg->GetAuxHeaderTable(0, "," CRLF_STR, ": ");
         int count = bmsg->GetAuxHeaderCount();
         if (noofstores) *noofstores = count;
         if (store_ids) *store_ids = (char **)apr_palloc(state.mp, sizeof(char *)*count);
         if (pct_broker2compute) *pct_broker2compute = (float *)apr_palloc(state.mp, sizeof(float)*count);
         if (pct_complete) *pct_complete = (float *)apr_palloc(state.mp, sizeof(float)*count);
         if (pct_compute2broker) *pct_compute2broker = (float *)apr_palloc(state.mp, sizeof(float)*count);
         int resultno = 0;
         const char *val;
         char *remainder = NULL;
         char *ttoken = apr_strtok(streamcopy, "," CRLF_STR, &remainder);
         while (ttoken != NULL)
         {
            if (resultno >= count)
            {
               errprint("broken: too many lines found");
               retval = ACQUIRE_QUERY_ALL_WORK__PARSE_ERROR;
               break;
            }
            char *remainder2 = NULL;
            char *store_id = apr_strtok(ttoken, ": ", &remainder2);
            const char *pthou_data = apr_table_get(aux_tbl, store_id);
            int pthou_broker2compute = 0;
            int pthou_complete = 0;
            int pthou_compute2broker = 0;
            sscanf(pthou_data, "PTHOU%d.%d.%d", &pthou_broker2compute, &pthou_complete, &pthou_compute2broker);
            if (pct_broker2compute) (*pct_broker2compute)[resultno] = pthou_broker2compute / 10.;
            if (pct_complete) (*pct_complete)[resultno] = pthou_complete / 10.;
            if (pct_compute2broker) (*pct_compute2broker)[resultno] = pthou_compute2broker / 10.;
            if (store_ids) (*store_ids)[resultno] = (char *)apr_pstrdup(state.mp, store_id);
            ttoken = apr_strtok(NULL, "," CRLF_STR, &remainder);
            resultno++;
         }
         retval = ACQUIRE_QUERY_ALL_WORK__SUCCESS;
      }
      if (strcmp(bmsg->GetType(), "ALLWORK") == 0)
      {
         const apr_table_t *htable = bmsg->GetHeaderTable();
         const char *quotasize = apr_table_get(htable, "QuotaBytes");
         if (quotasize && quota)
         {
            apr_off_t ssize = 0;
            sscanf(quotasize, "%" APR_OFF_T_FMT, &ssize);
            *quota = ssize;
         }
      }
   }
   delete bmsg;
   apr_pool_destroy(mp);
   return retval;
}

int AcquireAddSSHHostToPool(const char *username_at_remotemachine, const char *password, char **list_of_failed_hosts)
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   char *userkeyname = (char *)StringUtils::varprintf((state.registration_state == ACQUIRE_USER_REGISTRATION_STATE__FINAL) ? "%s" : ".%s", latest_keyname_of_user); 
   bmsg->SetSender(userkeyname);
   free(userkeyname);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(state.operational_brokerkeyname);
   bmsg->AddTypeHeaderNOCRLF("ADD_SSH_HOST", state.username);
   char *tmp_remotemach = (char *)malloc(sizeof(char)*(strlen(username_at_remotemachine)+1));
   strcpy(tmp_remotemach, username_at_remotemachine);
   char *remainder = NULL;
   char *username = apr_strtok(tmp_remotemach, "@", &remainder);
   char *hostname = apr_strtok(NULL, "@", &remainder);
   bmsg->AddVerbatimPayloadString(username);
   bmsg->AddVerbatimPayloadString(hostname);
   if (password)
   {
      bmsg->AddVerbatimPayloadString(password);
      bmsg->AddVerbatimPayloadString((state.keydb)->WIPpassword);
   }
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   int retval = -1;
   if ((bmsg->GetProcessee()) && (bmsg->GetType()))
   {
      if (strcmp(bmsg->GetProcessee(), "DENIED") == 0) retval = 0;
      if (strcmp(bmsg->GetType(), "SSH_SUCCESS") == 0) retval = ACQUIRE_USER_ADD_SSH__SSH_SUCCESS;
      if (strcmp(bmsg->GetProcessee(), "FINGERPRINT_UNTRUSTED") == 0) retval = ACQUIRE_USER_ADD_SSH__FINGERPRINT_UNTRUSTED;
      if (strcmp(bmsg->GetProcessee(), "ROUTE_FAILED") == 0) retval = ACQUIRE_USER_ADD_SSH__SSH_ROUTING_FAILED;
   }
   if (((retval == ACQUIRE_USER_ADD_SSH__FINGERPRINT_UNTRUSTED) ||
        (retval == ACQUIRE_USER_ADD_SSH__SSH_ROUTING_FAILED)) &&
        (list_of_failed_hosts))
   {
      *list_of_failed_hosts =
         (bmsg->GetHowManyPayloads() == 1) ?
            (char *)apr_pstrdup(mp, bmsg->GetPayload(0)) :
            NULL;
   }
   delete bmsg;
   apr_pool_destroy(mp);
   return retval;
}

const char *AcquireListOfAccessibleClusters()
{
   if (AcquireClientStateIsIncomplete()) return 0;
   apr_pool_t *mp;
   apr_pool_create(&mp, (state.apr_iface)->getGlobalPool());
   BrokerMessage *bmsg = new BrokerMessage(mp);
   char *latest_keyname_of_user = (state.keydb)->GetLatestPrivateKeyNameOfOwner(state.username);
   bmsg->SetSender(latest_keyname_of_user);
   free(latest_keyname_of_user);
   bmsg->SetRecipient(state.operational_brokerkeyname);
   bmsg->AddTypeHeaderNOCRLF("GET_CLUSTERS", state.username);
   Broker4Client *broker4 = new Broker4Client(state.hostname, state.portno, mp);
   broker4->MessageExchange(bmsg, state.keydb);
   delete broker4;
   if ((bmsg->GetProcessee()) && (bmsg->GetType()))
   {
      if ((strcmp(bmsg->GetType(), "ACCESSIBLE_CLUSTERS") == 0) && (bmsg->IsPayloadCount(1)))
      {
         char *accessible = (char *)apr_pstrdup(state.mp, bmsg->GetPayload(0));
         delete bmsg;
         apr_pool_destroy(mp);
         return accessible;
      }
   }
   delete bmsg;
   apr_pool_destroy(mp);
   return NULL;
}

int AcquireClientIsInitialised()
{
   return isInitialised;
}
